
#include "writebuffer.h"

#include "constants.h"



using namespace std;


WriteQueue * WriteQueue::instance = 0;


WriteQueue * WriteQueue::self()
{
    if ( !instance )
        instance = new WriteQueue();

    return instance;
}


WriteQueue::WriteQueue()
  : m_lastResult( 0 ),
    m_retries( replicaMaxRetries ),
    m_mutex(),
    m_queue()
{
    pthread_mutex_init( &m_mutex, NULL );
    pthread_mutex_init( &m_mutex, NULL );
    pthread_cond_init( &m_cond, NULL );
}


int WriteQueue::submitMessage( pthread_t thread, const std::string& message )
{
    pthread_mutex_lock( &m_mutex );
    m_queue.push_back( make_pair( thread, message ) );

    do
    {
        pthread_cond_wait( &m_cond, &m_mutex );
    }
    while ( containsThread( thread ) );

    int result =  m_lastResult;
    pthread_mutex_unlock( &m_mutex );

    return result;
}


bool WriteQueue::hasMessageWaiting() const
{
    pthread_mutex_lock( &m_mutex );
    bool result = !m_queue.empty();
    pthread_mutex_unlock( &m_mutex );
    return result;
}


string WriteQueue::currentMessage() const
{
    pthread_mutex_lock( &m_mutex );
    string result = m_queue.front().second;
    pthread_mutex_unlock( &m_mutex );
    return result;
}


void WriteQueue::doneWithMessage( int result )
{
    pthread_mutex_lock( &m_mutex );
    if ( result < 0 && m_retries > 0 )
    {
        --m_retries;
    }
    else
    {
        m_lastResult = result;
        m_queue.pop_front();
        m_retries = replicaMaxRetries;
        pthread_cond_broadcast( &m_cond );
    }
    pthread_mutex_unlock( &m_mutex );

}


bool WriteQueue::containsThread( pthread_t thread ) const
{
    bool result = false;
    deque<pair<pthread_t,string> >::const_iterator it = m_queue.begin();
    for ( ; it != m_queue.end(); ++it )
    {
        if ( thread == (*it).first )
        {
            result = true;
            break;
        }
    }

    return result;
}
